package sink;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Date 2021/6/17 15:59
 * Desc TODO
 */
public class JDBCSinkMySQL {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStreamSource<Student> source = env.fromElements(new Student(null, "JackMa", 42));
        //3.将数据通过 jdbc 插入到 mysql 数据库
        source.addSink(JdbcSink.sink(
                // 输入 SQL 执行插入SQL 语句
                "INSERT INTO t_student(id,name,age) values (null,?,?)",
                    // 执行插入的赋值
                    (ps, student) -> {
                        ps.setString(1,student.name);
                        ps.setInt(2,student.age);
                    },
                //构造器
                // 执行的选项 设置批处理大小等参数
                JdbcExecutionOptions.builder()
                        .withBatchSize(1000)
                        .withBatchIntervalMs(200)
                        .withMaxRetries(5)
                        .build(),
                //4.参数配置 连接参数
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://192.168.88.163:3306/bigdata?useSSL=false")
                        .withUsername("root")
                        .withPassword("123456")
                        .withDriverName("com.mysql.jdbc.Driver")
                        .build()));
        //5.执行环境
        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }
}
